package raft

//
// this is an outline of the API that raft must expose to
// the service (or tester). see comments below for
// each of these functions for more details.
//
// rf = Make(...)
//   create a new Raft server.
// rf.Start(command interface{}) (index, term, isleader)
//   start agreement on a new log entry
// rf.GetState() (term, isLeader)
//   ask a Raft for its current term, and whether it thinks it is leader
// ApplyMsg
//   each time a new entry is committed to the log, each Raft peer
//   should send an ApplyMsg to the service (or tester)
//   in the same server.
//

import (
	//	"bytes"

	"bytes"
	"fmt"
	"log"
	"math/rand"
	"reflect"
	"strconv"
	"sync"
	"sync/atomic"
	"time"

	//	"6.824/labgob"

	"6.824/labgob"
	"6.824/labrpc"
)

//
// as each Raft peer becomes aware that successive log entries are
// committed, the peer should send an ApplyMsg to the service (or
// tester) on the same server, via the applyCh passed to Make(). set
// CommandValid to true to indicate that the ApplyMsg contains a newly
// committed log entry.
//
// in part 2D you'll want to send other kinds of messages (e.g.,
// snapshots) on the applyCh, but set CommandValid to false for these
// other uses.
//
type ApplyMsg struct {
	CommandValid bool
	Command      interface{}
	CommandIndex int

	// For 2D:
	SnapshotValid bool
	Snapshot      []byte
	SnapshotTerm  int
	SnapshotIndex int
}
type LogLevel string

const (
	APPLYLOG         LogLevel = "apply"
	APPENDLOG        LogLevel = "append"
	ELECTIONLOG      LogLevel = "election"
	HEARTBEATLOG     LogLevel = "heartbeat"
	REQUESTVOTELOG   LogLevel = "requetsvote"
	APPENDENTRIESLOG LogLevel = "appendentries"
	ROLECHANGE       LogLevel = "roleChange"
	STATUSLOG        LogLevel = "status"
)

type Role int

const (
	Follower   Role = 0
	Leader     Role = 1
	Candidater Role = 2
)
const (
	// 不同时间配合产生不同结果
	HeartBeatTimeout = 90
	// 时间间隔边长, 减少选举次数
	BaseElectionTimeout = 200
	// 确保一个选举周期之内完成一个选举
	RPCTimeout = 200
)

//
// A Go object implementing a single Raft peer.
//
type Raft struct {
	mu        sync.Mutex          // Lock to protect shared access to this peer's state
	peers     []*labrpc.ClientEnd // RPC end points of all peers
	persister *Persister          // Object to hold this peer's persisted state
	me        int                 // this peer's index into peers[]
	dead      int32               // set by Kill()

	// Your data here (2A, 2B, 2C).
	// Look at the paper's Figure 2 for a description of what
	// state a Raft server must maintain.
	electionTimeOut *time.Ticker
	role            Role
	term            int

	voteFor     int
	logs        []*Entry
	commitIndex int
	lastApplied int
	nextIndex   []int
	matchIndex  []int

	grantVote int
	voteChan  chan bool

	applyCond   *sync.Cond
	replicaCond []*sync.Cond

	applyCh chan ApplyMsg
	// 进入time.Ticker, 需要基于这个字段判断是否进入选举周期

	beginElection bool
	// 禁止打印log
	printLog   bool
	replicaRPC []int
}

type Entry struct {
	Command interface{}
	Index   int
	Term    int
}

func (rf *Raft) changePrintLOG(b bool) {
	rf.mu.Lock()
	defer rf.mu.Unlock()
	rf.printLog = b
}

func (rf *Raft) roleToStr() string {
	r := ""
	switch {
	case rf.role == Candidater:
		r = "candidate"
	case rf.role == Follower:
		r = "follower"
	case rf.role == Leader:
		r = "leader"
	}
	return r
}

// return currentTerm and whether this server
// believes it is the leader.
func (rf *Raft) GetState() (int, bool) {

	var term int
	var isleader bool
	// Your code here (2A).
	rf.mu.Lock()
	defer rf.mu.Unlock()
	// log.Printf("node %d role %s grant %d", rf.me, rf.roleToStr(rf.role), rf.grantVote)
	isleader = rf.role == Leader
	term = rf.term
	rf.logRaft(term, -1, STATUSLOG, fmt.Sprintf("%t", isleader))
	return term, isleader
}

//
// save Raft's persistent state to stable storage,
// where it can later be retrieved after a crash and restart.
// see paper's Figure 2 for a description of what should be persistent.
//
func (rf *Raft) persist() {
	// Your code here (2C).
	// Example:
	w := new(bytes.Buffer)
	e := labgob.NewEncoder(w)
	e.Encode(rf.term)
	e.Encode(rf.voteFor)
	e.Encode(rf.logs)
	e.Encode(rf.lastApplied)
	data := w.Bytes()
	rf.persister.SaveRaftState(data)
}

//
// restore previously persisted state.
//
func (rf *Raft) readPersist(data []byte) {
	if data == nil || len(data) < 1 { // bootstrap without any state?
		return
	}
	// Your code here (2C).
	// Example:
	rf.mu.Lock()
	defer rf.mu.Unlock()
	r := bytes.NewBuffer(data)
	d := labgob.NewDecoder(r)
	var term int
	var voteFor int
	var logs []*Entry
	var lastApplied int

	if err := d.Decode(&term); err != nil {
		log.Fatal(err)
	} else {
		rf.term = term
	}
	if err := d.Decode(&voteFor); err != nil {
		log.Fatal(err)
	} else {
		rf.voteFor = voteFor
	}

	if err := d.Decode(&logs); err != nil {
		log.Fatal(err)
	} else {
		rf.logs = logs
	}
	if err := d.Decode(&lastApplied); err != nil {
		log.Fatal(err)
	} else {
		rf.lastApplied = lastApplied
	}

}

//
// A service wants to switch to snapshot.  Only do so if Raft hasn't
// have more recent info since it communicate the snapshot on applyCh.
//
func (rf *Raft) CondInstallSnapshot(lastIncludedTerm int, lastIncludedIndex int, snapshot []byte) bool {

	// Your code here (2D).

	return true
}

// the service says it has created a snapshot that has
// all info up to and including index. this means the
// service no longer needs the log through (and including)
// that index. Raft should now trim its log as much as possible.
func (rf *Raft) Snapshot(index int, snapshot []byte) {
	// Your code here (2D).

}

//
// example RequestVote RPC arguments structure.
// field names must start with capital letters!
//
type RequestVoteArgs struct {
	// Your data here (2A, 2B).
	Term         int
	CandidaterID int
	LastLogIndex int
	LastLogTerm  int
}

//
// example RequestVote RPC reply structure.
// field names must start with capital letters!
//
type RequestVoteReply struct {
	// Your data here (2A).
	Term        int
	VoteGranted bool
}

//
// example RequestVote RPC handler.
//
func (rf *Raft) RequestVote(args *RequestVoteArgs, reply *RequestVoteReply) {
	// Your code here (2A, 2B).
	rf.mu.Lock()
	defer rf.mu.Unlock()

	rf.beginElection = false
	rf.DownGrade(args.Term)
	reply.Term = rf.term
	// 1. Reply false if term < currentTerm (§5.1)
	if args.Term < rf.term {
		reply.VoteGranted = false
		rf.logRaft(args.Term, args.CandidaterID, REQUESTVOTELOG, fmt.Sprintf("%d->%d, status 1, false ", args.CandidaterID, rf.me))

		return
	}

	// 2. If votedFor is null or candidateId
	if rf.voteFor == -1 || rf.voteFor == args.CandidaterID {
		last := rf.LastEntry()
		// 3. candidate’s log is at
		// least as up-to-date as receiver’s log
		// If the logs have last entries with different terms, then
		// the log with the later term is more up-to-date. If the logs
		// end with the same term, then whichever log is longer is
		// more up-to-date.

		if last.Term < args.LastLogTerm || (last.Term == args.LastLogTerm && last.Index <= args.LastLogIndex) {
			t := rf.randElectionTimeout()
			rf.electionTimeOut.Reset(t)
			reply.VoteGranted = true
			rf.voteFor = args.CandidaterID
			rf.logRaft(args.Term, args.CandidaterID, REQUESTVOTELOG, fmt.Sprintf("%d->%d, status 2, true, %d reset: %d", args.CandidaterID, rf.me, rf.me, t))
			rf.persist()

			return
		}

	}
	rf.logRaft(args.Term, args.CandidaterID, REQUESTVOTELOG, fmt.Sprintf("%d->%d status 3,voteFor %d, role %s,  false", args.CandidaterID, rf.me, rf.voteFor, rf.roleToStr()))

}

//
// example code to send a RequestVote RPC to a server.
// server is the index of the target server in rf.peers[].
// expects RPC arguments in args.
// fills in *reply with RPC reply, so caller should
// pass &reply.
// the types of the args and reply passed to Call() must be
// the same as the types of the arguments declared in the
// handler function (including whether they are pointers).
//
// The labrpc package simulates a lossy network, in which servers
// may be unreachable, and in which requests and replies may be lost.
// Call() sends a request and waits for a reply. If a reply arrives
// within a timeout interval, Call() returns true; otherwise
// Call() returns false. Thus Call() may not return for a while.
// A false return can be caused by a dead server, a live server that
// can't be reached, a lost request, or a lost reply.
//
// Call() is guaranteed to return (perhaps after a delay) *except* if the
// handler function on the server side does not return.  Thus there
// is no need to implement your own timeouts around Call().
//
// look at the comments in ../labrpc/labrpc.go for more details.
//
// if you're having trouble getting RPC to work, check that you've
// capitalized all field names in structs passed over RPC, and
// that the caller passes the address of the reply struct with &, not
// the struct itself.
//
func (rf *Raft) sendRequestVote(server int, args *RequestVoteArgs, reply *RequestVoteReply) bool {

	ok := rf.peers[server].Call("Raft.RequestVote", args, reply)

	return ok
}

type AppendEntriesArgs struct {
	// Your data here (2A, 2B).
	Term         int
	LeaderID     int
	PrevLogIndex int
	PrevLogTerm  int
	Entries      []*Entry
	LeaderCommit int
}

//
// example RequestVote RPC reply structure.
// field names must start with capital letters!
//
type AppendEntriesReply struct {
	// Your data here (2A).
	Term    int
	Success bool
}

func min(x, y int) int {
	if x < y {
		return x
	}
	return y
}
func (rf *Raft) AppendEntries(args *AppendEntriesArgs, reply *AppendEntriesReply) {
	// Your code here (2A, 2B).

	rf.mu.Lock()
	defer rf.mu.Unlock()
	t := rf.randElectionTimeout()

	rf.electionTimeOut.Reset(t)
	rf.beginElection = false

	// args.Entries[len(args.Entries)-1].Index

	rf.DownGrade(args.Term)
	reply.Term = rf.term
	// 1. Reply false if term < currentTerm (§5.1)
	if args.Term < rf.term {
		rf.logRaft(args.Term, args.LeaderID, APPENDENTRIESLOG, fmt.Sprintf("%d->%d Reply false if term < currentTerm (§5.1),requestTerm %d, currentTerm: %d,  %d reset: %d", args.LeaderID, rf.me, args.Term, rf.term, rf.me, t))

		reply.Success = false
		reply.Term = rf.term
		return
	}

	// 2. Reply false if log doesn’t contain an entry at prevLogIndex
	// whose term matches prevLogTerm (§5.3)
	if len(rf.logs) <= args.PrevLogIndex {
		rf.logRaft(args.Term, args.LeaderID, APPENDENTRIESLOG, fmt.Sprintf("%d->%d Reply false if log doesn’t contain an entry at prevLogIndex, prev: term%d, index, %d;  entries: %d,  %d reset: %d", args.LeaderID, rf.me, args.PrevLogTerm, args.PrevLogIndex, len(args.Entries), rf.me, t))
		reply.Success = false
		return
	}

	if rf.logs[args.PrevLogIndex].Term != args.PrevLogTerm {
		reply.Success = false
		// 3. If an existing entry conflicts with a new one (same index
		// 	but different terms), delete the existing entry and all that
		// 	follow it (§5.3)
		rf.logs = rf.logs[:args.PrevLogIndex]
		rf.logRaft(args.Term, args.LeaderID, APPENDENTRIESLOG, fmt.Sprintf("%d->%d  If an existing entry conflicts with a new one ,commitIndex %d, entries: %d,  %d reset: %d", args.LeaderID, rf.me, rf.commitIndex, len(args.Entries), rf.me, t))

		return
	}

	// 4. Append any new entries not already in the log
	if len(args.Entries) > 0 {
		rf.logs = rf.logs[:args.PrevLogIndex+1]

		for i := 0; i < len(args.Entries); i++ {
			tmp := *args.Entries[i]
			rf.logs = append(rf.logs, &tmp)

		}
		rf.logRaft(args.Term, args.LeaderID, APPENDENTRIESLOG, fmt.Sprintf("%d->%d  Append any new entries not already in the log,commitIndex %d, entries: %d, logs: %d, prevIndex: %d, %d reset: %d", args.LeaderID, rf.me, rf.commitIndex, len(args.Entries), len(rf.logs), args.PrevLogIndex, rf.me, t))
		rf.persist()
		reply.Success = true
	}

	// 5. If leaderCommit > commitIndex, set commitIndex =
	// min(leaderCommit, index of last new entry)
	if args.LeaderCommit > rf.commitIndex {
		c := min(args.LeaderCommit, rf.lastEntry().Index)
		rf.logRaft(args.Term, args.LeaderID, APPENDENTRIESLOG, fmt.Sprintf("%d->%d  If leaderCommit > commitIndex,ocommitIndex %d, ncommitIndex %d, entries: %d,  %d reset: %d", args.LeaderID, rf.me, rf.commitIndex, c, len(args.Entries), rf.me, t))

		rf.commitIndex = c

		rf.applyCond.Signal()
	}
}

func (rf *Raft) sendAppendEntries(server int, args *AppendEntriesArgs, reply *AppendEntriesReply) bool {

	ok := rf.peers[server].Call("Raft.AppendEntries", args, reply)

	return ok
}

func (rf *Raft) LastEntry() *Entry {

	return rf.logs[len(rf.logs)-1]
}

//
// the service using Raft (e.g. a k/v server) wants to start
// agreement on the next command to be appended to Raft's log. if this
// server isn't the leader, returns false. otherwise start the
// agreement and return immediately. there is no guarantee that this
// command will ever be committed to the Raft log, since the leader
// may fail or lose an election. even if the Raft instance has been killed,
// this function should return gracefully.
//
// the first return value is the index that the command will appear at
// if it's ever committed. the second return value is the current
// term. the third return value is true if this server believes it is
// the leader.
//
func (rf *Raft) Start(command interface{}) (int, int, bool) {
	index := 0
	term := 0
	isLeader := false
	// Your code here (2B).
	rf.mu.Lock()
	role := rf.role

	if role == Leader {
		last := rf.lastEntry()
		index = last.Index + 1
		isLeader = true
		term = rf.term
		// 1. appends the command to its log as a new entry
		entry := &Entry{
			Command: command,
			Index:   index,
			Term:    term,
		}
		rf.logs = append(rf.logs, entry)
		rf.logRaft(term, rf.me, APPENDLOG, fmt.Sprintf("start command: %v, lastIndex %d", command, rf.LastEntry().Index))
		rf.persist()
		// 2. issues AppendEntries RPCs in parallel to each of the other
		// servers to replicate the
		for i, _ := range rf.peers {
			if i != rf.me {
				rf.replicaCond[i].Signal()
			}
		}
		rf.mu.Unlock()
	} else {

		rf.mu.Unlock()
		return index, term, isLeader
	}
	c := 0
	for {
		rf.mu.Lock()
		c++
		if rf.role != Leader {
			isLeader = false
			rf.logRaft(term, rf.me, APPENDLOG, fmt.Sprintf("end command become follow : %v, lastIndex %d", command, rf.LastEntry().Index))

			rf.mu.Unlock()

			return index, term, isLeader

		}
		if rf.replicaIndexOverMajority(index) {
			// the leader applies the entry to its state machine
			rf.applyCond.Signal()
			rf.logRaft(term, rf.me, APPENDLOG, fmt.Sprintf("end command: %v, lastIndex %d", command, rf.LastEntry().Index))

			rf.mu.Unlock()
			// returns the result of that execution to the client

			return index, term, isLeader

		}
		if c == 5 {
			rf.logRaft(term, rf.me, APPENDLOG, fmt.Sprintf("end command: %v timeout, lastIndex %d", command, rf.LastEntry().Index))

			rf.mu.Unlock()

			return index, term, isLeader
		}
		rf.mu.Unlock()

		time.Sleep(time.Millisecond * 30)
	}

}

//
// the tester doesn't halt goroutines created by Raft after each test,
// but it does call the Kill() method. your code can use killed() to
// check whether Kill() has been called. the use of atomic avoids the
// need for a lock.
//
// the issue is that long-running goroutines use memory and may chew
// up CPU time, perhaps causing later tests to fail and generating
// confusing debug output. any goroutine with a long-running loop
// should call killed() to check whether it should stop.
//
func (rf *Raft) Kill() {
	atomic.StoreInt32(&rf.dead, 1)
	// Your code here, if desired.

}

func (rf *Raft) killed() bool {
	z := atomic.LoadInt32(&rf.dead)
	return z == 1
}
func (rf *Raft) randElectionTimeout() time.Duration {
	rand.Seed(time.Now().Unix() + int64(rf.me))
	t := time.Millisecond*BaseElectionTimeout + time.Duration(rand.Intn(10))*time.Millisecond*30

	return t
}

func (rf *Raft) broadcase() {
	rf.mu.Lock()
	defer rf.mu.Unlock()

	for peer := 0; peer < len(rf.peers); peer++ {
		if rf.role == Leader {
			if peer != rf.me {
				nextIndex := rf.nextIndex[peer]
				prev := rf.logs[nextIndex-1]
				request := &AppendEntriesArgs{
					Term:         rf.term,
					LeaderID:     rf.me,
					PrevLogIndex: prev.Index,
					PrevLogTerm:  prev.Term,
					Entries:      []*Entry{},
					LeaderCommit: rf.commitIndex,
				}
				response := &AppendEntriesReply{}
				go func(peer int, request *AppendEntriesArgs, response *AppendEntriesReply) {
					rf.sendAppendEntries(peer, request, response)

					rf.mu.Lock()
					rf.DownGrade(response.Term)
					rf.logRaft(rf.term, rf.me, HEARTBEATLOG, fmt.Sprintf("%d->%d, heardbeat result, response term %d, term %d, role %s", request.LeaderID, peer, response.Term, rf.term, rf.roleToStr()))
					rf.mu.Unlock()

				}(peer, request, response)
			}

		}
	}

}

func (rf *Raft) marjority() int {
	// 4  2 + 1 = 3
	// 3  1 + 1 = 2
	// 7  3 + 1 = 4
	return int(len(rf.peers)/2) + 1
}

func (rf *Raft) firstTermByTerm(term int) Entry {
	entry := Entry{}
	for i := 0; i < len(rf.logs); i++ {
		if rf.logs[i].Term == term {
			entry = *rf.logs[i]
			return entry
		}
	}
	return entry
}

func commandToString(command interface{}) string {
	c := ""
	if command != nil {

		switch v := reflect.ValueOf(command); v.Kind() {
		case reflect.String:
			c = v.String()
		case reflect.Int, reflect.Int8, reflect.Int16, reflect.Int32, reflect.Int64:
			c = strconv.Itoa(int(v.Int()))

		default:
			fmt.Printf("unhandled kind %s", v.Kind())
		}
	}
	return c
}

func EntriesToString(entries []*Entry) string {
	s := ""
	for _, v := range entries {

		s = s + fmt.Sprintf("#%s", commandToString(v.Command))

	}
	return s
}

type logWriter struct {
}

func (writer logWriter) Write(bytes []byte) (int, error) {
	return fmt.Print(time.Now().UTC().Format("2006-02-01 15:04:05.000") + " [DEBUG]" + string(bytes))
}
func (rf *Raft) logRaft(term, leader int, level LogLevel, info ...string) {
	// term-leader-level
	if !rf.killed() && rf.printLog {
		log.SetFlags(0)
		log.SetOutput(new(logWriter))
		base := fmt.Sprintf("[%d-%d-%s]", term, leader, string(level))
		filterpprint := func(term int, level LogLevel, info ...string) {
			s := ""

			for _, v := range info {
				s += v
			}
			// 过滤
			switch level {
			case APPLYLOG:
				return
			case APPENDLOG:

				return
			case ELECTIONLOG:
				return
			case HEARTBEATLOG:
				return
			case REQUESTVOTELOG:
				return
			case APPENDENTRIESLOG:
				return
			case ROLECHANGE:
				return
			case STATUSLOG:
				return
			}
			log.Printf("%s %s \n", base, s)

		}
		filterpprint(term, level, info...)
	}

}

func (rf *Raft) DownGrade(requestTerm int) {
	if requestTerm > rf.term {
		rf.logRaft(rf.term, -1, ROLECHANGE, fmt.Sprintf("%d to follow, term %d->%d", rf.me, rf.term, requestTerm))
		rf.role = Follower
		rf.term = requestTerm
		// rf.electionTimeOut.Reset(rf.randElectionTimeout())
		rf.voteFor = -1

	}

}

func (rf *Raft) doVoteForPeer(peer int, request *RequestVoteArgs, response *RequestVoteReply) {
	rf.sendRequestVote(peer, request, response)

	rf.mu.Lock()
	defer rf.mu.Unlock()

	rf.DownGrade(response.Term)
	if rf.role == Candidater && rf.term == request.Term {
		if response.VoteGranted {
			if rf.grantVote < rf.marjority() {
				rf.grantVote++
				if rf.grantVote >= rf.marjority() {
					go rf.broadcase()
					rf.ininNextIndex()
					rf.beginElection = false
					rf.role = Leader
					rf.logRaft(request.Term, rf.me, ELECTIONLOG, "become leader")
				}
			}
		}
	}
}

func (rf *Raft) voteForPeer(term, peer int) {

	last := rf.lastEntry()
	response := &RequestVoteReply{}
	request := &RequestVoteArgs{
		Term:         term,
		CandidaterID: rf.me,
		LastLogIndex: last.Index,
		LastLogTerm:  last.Term,
	}

	go rf.doVoteForPeer(peer, request, response)

}

// The ticker go routine starts a new election if this peer hasn't received
// heartbeats recently.
func (rf *Raft) ticker() {

	for !rf.killed() {

		// Your code here to check if a leader election should
		// be started and to randomize sleeping time using
		// time.Sleep().

		select {
		case <-rf.electionTimeOut.C:
			rf.mu.Lock()
			t := rf.randElectionTimeout()
			if rf.beginElection {
				rf.term = rf.term + 1
				term := rf.term

				rf.role = Candidater
				rf.voteFor = rf.me
				rf.grantVote = 1
				rf.persist()
				for peer := 0; peer < len(rf.peers); peer++ {
					if peer != rf.me {

						go rf.voteForPeer(term, peer)
					}

				}

				if rf.role == Leader {
					rf.beginElection = false
				}

				rf.logRaft(term, rf.me, ELECTIONLOG, fmt.Sprintf("%d %s start election, reset: %d", rf.me, rf.roleToStr(), t))

			} else {
				rf.beginElection = true
				rf.logRaft(rf.term, rf.me, ELECTIONLOG, fmt.Sprintf("%d %s not start election", rf.me, rf.roleToStr()))

			}
			rf.electionTimeOut.Reset(t)
			rf.mu.Unlock()
		default:
			rf.mu.Lock()
			if rf.role == Leader {
				rf.electionTimeOut.Reset(rf.randElectionTimeout())
				go rf.broadcase()
				rf.beginElection = false
				rf.logRaft(rf.term, rf.me, HEARTBEATLOG, fmt.Sprintf("%d  %s heartbeat", rf.me, rf.roleToStr()))
			}
			rf.mu.Unlock()
			time.Sleep(HeartBeatTimeout * time.Millisecond)

		}
	}

}
func (rf *Raft) applyDone() bool {
	// 确定commitIndex
	rf.deduceCommitIndex()
	done := false
	last := rf.lastEntry()
	if last.Term == rf.term {
		done = rf.commitIndex > rf.lastApplied
	}

	return done
}

func (rf *Raft) replicaDone(peer int) bool {

	done := false
	// role == leader
	if rf.role != Leader {
		return done
	}
	last := rf.lastEntry()

	if rf.nextIndex[peer] <= last.Index && rf.replicaRPC[peer] == 0 {

		done = true
	}

	return done
}

func (rf *Raft) replicaToPeer(peer int) {

	for {

		rf.replicaCond[peer].L.Lock()

		for !rf.replicaDone(peer) {
			rf.replicaCond[peer].Wait()
		}
		rf.replicaRPC[peer] = 1
		go rf.doReplicaToPeer(peer)
		rf.replicaCond[peer].L.Unlock()

	}
}

func (rf *Raft) doReplicaToPeer(peer int) {
	rf.mu.Lock()
	// 准备参数

	nextIndex := rf.nextIndex[peer]
	entries := rf.logs[nextIndex:]

	prev := rf.logs[nextIndex-1]
	args := &AppendEntriesArgs{
		Term:         rf.term,
		LeaderID:     rf.me,
		PrevLogIndex: prev.Index,
		PrevLogTerm:  prev.Term,
		Entries:      entries,
		LeaderCommit: rf.commitIndex,
	}
	if len(entries) == 0 {
		rf.mu.Unlock()
		return
	}
	reply := &AppendEntriesReply{}
	rf.logRaft(rf.term, rf.me, APPENDLOG, fmt.Sprintf("%d->%d start_replicas, nextIndex: %d, lastIndex: %d, matchIndex:%d, entries %d, rpccount %d", rf.me, peer, rf.nextIndex[peer], rf.lastEntry().Index, rf.matchIndex[peer], len(entries), rf.replicaRPC[peer]))

	rf.mu.Unlock()

	ok := rf.sendAppendEntries(peer, args, reply)

	rf.mu.Lock()
	defer rf.mu.Unlock()

	if ok {
		// If RPC request or response contains term T > currentTerm:
		// set currentTerm = T, convert to follower (§5.1)
		rf.DownGrade(reply.Term)
		if rf.role == Leader {
			// If successful: update nextIndex and matchIndex for
			// follower (§5.3)
			if reply.Success {
				rf.nextIndex[peer] = prev.Index + len(entries) + 1
				rf.matchIndex[peer] = prev.Index + len(entries)
				rf.logRaft(rf.term, rf.me, APPENDLOG, fmt.Sprintf("%d->%d append  succeed, nextIndex: %d, lastIndex: %d, matchIndex:%d ", rf.me, peer, rf.nextIndex[peer], rf.lastEntry().Index, rf.matchIndex[peer]))

				rf.applyCond.Signal()

			} else {
				// If AppendEntries fails because of log inconsistency:
				// decrement nextIndex and retry (§5.3)
				rf.logRaft(rf.term, rf.me, APPENDLOG, fmt.Sprintf("%d->%d append  log inconsistency, nextIndex: %d, lastIndex: %d, matchIndex:%d ", rf.me, peer, rf.nextIndex[peer], rf.lastEntry().Index, rf.matchIndex[peer]))

				if rf.nextIndex[peer] > 1 {
					// 优化
					e := rf.firstTermByTerm(rf.logs[args.PrevLogIndex].Term)
					i := e.Index
					switch {
					case i == 1:
						rf.nextIndex[peer] = 1
					case i > 1:
						rf.nextIndex[peer] = rf.firstTermByTerm(rf.logs[i].Term).Index
					}
					if i > 1 {
						i = i - 1
					}

					// 原始
					// rf.nextIndex[peer] = rf.nextIndex[peer] - 1
					rf.replicaCond[peer].Signal()
				}
			}
		}

	} else {
		// follower crash or run slowly, 重新发送
		rf.logRaft(rf.term, rf.me, APPENDLOG, fmt.Sprintf("%d->%d append  follower crash or run slowly, nextIndex: %d, lastIndex: %d, matchIndex:%d ", rf.me, peer, rf.nextIndex[peer], rf.lastEntry().Index, rf.matchIndex[peer]))

		rf.replicaCond[peer].Signal()

	}
	rf.replicaRPC[peer] = 0
}

func (rf *Raft) replicaIndexOverMajority(index int) bool {
	num := 1

	for i := 0; i < len(rf.matchIndex); i++ {
		if i != rf.me {
			if rf.matchIndex[i] >= index {
				num++
				if num >= rf.marjority() {
					return true
				}
			}
		}

	}
	return false
}

func (rf *Raft) deduceCommitIndex() {
	// If there exists an N such that N > commitIndex, a majority
	// of matchIndex[i] ≥ N, and log[N].term == currentTerm:
	// set commitIndex = N (§5.3, §5.4).

	if rf.role == Leader {

		for i := len(rf.logs) - 1; i > rf.commitIndex; i-- {
			if rf.replicaIndexOverMajority(i) && rf.logs[i].Term == rf.term {
				rf.commitIndex = i
				return
			}
		}
	}
}

func (rf *Raft) apply() {
	for {

		rf.applyCond.L.Lock()

		for !rf.applyDone() {

			rf.applyCond.Wait()
		}
		tmp := 0
		// If commitIndex > lastApplied: increment lastApplied, apply
		// log[lastApplied] to state machine (§5.3)
		for i := rf.lastApplied; i < rf.commitIndex; i++ {
			tmp = i + 1
			rf.logRaft(rf.term, rf.me, APPLYLOG, fmt.Sprintf("lastApplies: %d, commitIndex: %d, commit %d , lastIndex %d, entries %v", rf.lastApplied, rf.commitIndex, tmp, rf.lastEntry().Index, rf.logs[tmp].Command))

			rf.applyCh <- ApplyMsg{
				CommandValid: true,
				Command:      rf.logs[tmp].Command,
				CommandIndex: rf.logs[tmp].Index,
			}
			rf.lastApplied = tmp
			rf.persist()
		}

		rf.applyCond.L.Unlock()
	}
}

func (rf *Raft) lastEntry() *Entry {
	return rf.logs[len(rf.logs)-1]
}

// Reinitialized after election， initialized to leader
// last log index + 1
func (rf *Raft) ininNextIndex() {
	last := rf.LastEntry()
	for i := 0; i < len(rf.peers); i++ {
		rf.nextIndex[i] = last.Index + 1
	}
}

//
// the service or tester wants to create a Raft server. the ports
// of all the Raft servers (including this one) are in peers[]. this
// server's port is peers[me]. all the servers' peers[] arrays
// have the same order. persister is a place for this server to
// save its persistent state, and also initially holds the most
// recent saved state, if any. applyCh is a channel on which the
// tester or service expects Raft to send ApplyMsg messages.
// Make() must return quickly, so it should start goroutines
// for any long-running work.
//
func Make(peers []*labrpc.ClientEnd, me int,
	persister *Persister, applyCh chan ApplyMsg) *Raft {
	logs := make([]*Entry, 0)
	logs = append(logs, &Entry{
		Command: nil,
		Index:   0,
		Term:    0,
	})

	rf := &Raft{}
	rf.peers = peers
	rf.persister = persister
	rf.me = me
	rf.logs = logs
	rf.commitIndex = 0
	rf.lastApplied = 0
	rf.nextIndex = make([]int, len(peers))
	rf.ininNextIndex()
	rf.matchIndex = make([]int, len(peers))
	// Your initialization code here (2A, 2B, 2C).
	rf.electionTimeOut = time.NewTicker(rf.randElectionTimeout())
	rf.role = 0
	rf.voteFor = -1
	rf.grantVote = 0
	rf.term = 0
	rf.replicaRPC = make([]int, len(peers))

	rf.voteChan = make(chan bool)
	rf.applyCond = sync.NewCond(&rf.mu)
	rf.replicaCond = make([]*sync.Cond, 0)

	rf.applyCh = applyCh
	rf.beginElection = true
	rf.printLog = true

	for i := 0; i < len(rf.peers); i++ {
		rf.replicaCond = append(rf.replicaCond, sync.NewCond(&rf.mu))
	}

	for i := 0; i < len(rf.peers); i++ {
		if i != rf.me {
			go rf.replicaToPeer(i)
		}

	}

	go rf.apply()

	// initialize from state persisted before a crash
	rf.readPersist(persister.ReadRaftState())

	// start ticker goroutine to start elections
	go rf.ticker()

	return rf
}
